package com.at.aggregate6;

import com.at.bean.WaterSensor;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * @author huangchao E-mail:fengquan8866@163.com
 * @version 创建时间：2024/9/25 11:36
 */
public class SimpleAggregateDemo2 {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        DataStreamSource<WaterSensor> sensorDS = env.fromElements(
                new WaterSensor("s1", 1L, 1),
                new WaterSensor("s1", 11L, 11),
                new WaterSensor("s2", 2L, 2),
                new WaterSensor("s3", 3L, 3)
        );

        KeyedStream<WaterSensor, String> sensorKS = sensorDS.keyBy(new KeySelector<WaterSensor, String>() {
            @Override
            public String getKey(WaterSensor value) throws Exception {
                return value.getId();
            }
        });

        /**
         * TODO 简单聚合算子
         * 1、keyby之后调用
         * 2、分组内的聚合：对同一个key的数据进行聚合
         */
        // 传位置索引的，适用于Tuple类型
//        SingleOutputStreamOperator<WaterSensor> rst = sensorKS.sum(3);
//        SingleOutputStreamOperator<WaterSensor> rst = sensorKS.sum("vc");

//        SingleOutputStreamOperator<WaterSensor> rst = sensorKS.min("vc");

//        SingleOutputStreamOperator<WaterSensor> rst = sensorKS.minBy("vc");

        /**
         * max/maxBy 的区别：
         *   max：只会取比较字段的最大值，非比较字段保留第一次的值
         *   maxby：取比较字段的最大值，同时非比较字段取 最大值这条跳狙的值
         */
//        SingleOutputStreamOperator<WaterSensor> rst = sensorKS.max("vc");
        SingleOutputStreamOperator<WaterSensor> rst = sensorKS.maxBy("vc");

        rst.print();

        env.execute();
    }
}
